交易额|成交额_实战SparkStream+Kafka+Redis实时计算商品销售额

篇首语：本文由编程笔记#小编为大家整理，主要介绍了实战SparkStream+Kafka+Redis实时计算商品销售额相关的知识，希望对你有一定的参考价值。

写在前面

2016年天猫双十一当天&＃xff0c;零点的倒计时话音未落&＃xff0c;52秒交易额冲破10亿。随后&＃xff0c;又迅速在0时6分28秒&＃xff0c;达到100亿&＃xff01;每一秒开猫大屏上的交易额都在刷新&＃xff0c;这种时实刷新的大屏看着感觉超爽。天猫这个大屏后面的技术应该是使用流计算&＃xff0c;阿里使用Java将Storm重写了&＃xff0c;叫JStrom(https://github.com/alibaba/jstorm)&＃xff0c;最近学习SparkStream和Kafka,可以简单模仿一下这个时实计算成交额的过程&＃xff0c;主要目的是实际运用这些技术&＃xff0c;也了解一下技术的运用场景&＃xff0c;加深对技术的理解。

实时计算模型

下图所示为通用SparkStream时实计算模型&＃xff0c;主要分为三部分

1. 数据源
   我们这里的数据源选用了Kafka&＃xff0c;关于Kafka的安装与使用说明可以参考这里https://kafkadoc.beanmr.com/

2. SparkStream计算
   SparkStream是实时计算的核心&＃xff0c;这们这里也是近时实计算&＃xff0c;选择一个时间窗口&＃xff0c;对时间窗口中的数据做离线计算。

3. 数据落地
   SparkStream算好的结果可以存HDFS/mysql/Redis等等&＃xff0c;我们这里对商品销售额计算过程有涉及累加&＃xff0c;所以选择了Redis

业务模型介绍

我们模仿一个电商系统&＃xff0c;每时每刻都有订单成交&＃xff0c;每一笔成交的数据以一个事件发送到Kafka中&＃xff0c;SparkStream每一分中从Kafka中读取一次数据&＃xff0c;计算一分钟内每个商品的销售额&＃xff0c;然而写入Redis&＃xff0c;并在Redis中累加每分钟的数据&＃xff0c;Redis中主要存三种结果数量&＃xff0c;从开始到当前总销售额、从开始到当前每个商品销售额、上一分钟每个商品的销售额

Kafka生产者&＃xff0c;模拟每时每刻订单交易

object OrderProducer
def main(args: Array[String]): Unit &＃61;
//Kafka参数设置
val topic &＃61; "order"
val brokers &＃61; "127.0.0.1:9092"
val props &＃61; new Properties()
props.put("metadata.broker.list", brokers)
props.put("serializer.class", "kafka.serializer.StringEncoder")
val kafkaConfig &＃61; new ProducerConfig(props)
//创建生产者
val producer &＃61; new Producer[String, String](kafkaConfig)
while (true)
//随机生成10以内ID
val id &＃61; Random.nextInt(10)
//创建订单成交事件
val event &＃61; new JSONObject();
//商品ID
event.put("id", id)
//商品成交价格
event.put("price", Random.nextInt(10000))
//发送信息
producer.send(new KeyedMessage[String, String](topic, event.toString))
println("Message sent: " &＃43; event)
//随机暂停一段时间
Thread.sleep(Random.nextInt(100))




生产者输出结果&＃xff1a;

Message sent: "price":3959,"id":6
Message sent: "price":1579,"id":0
Message sent: "price":857,"id":6
Message sent: "price":8440,"id":1
Message sent: "price":6873,"id":6
Message sent: "price":6202,"id":2
Message sent: "price":8403,"id":6
Message sent: "price":7866,"id":2
Message sent: "price":9441,"id":5
Message sent: "price":6880,"id":4
Message sent: "price":4572,"id":5
Message sent: "price":509,"id":3
Message sent: "price":7526,"id":0

上述代码主要模拟一家店铺有十件商品&＃xff0c;ID从0到9&＃xff0c;每隔一小段随机时间成交一单&＃xff0c;成交价格以分为单位&＃xff0c;每成交一笔就像Kafka中发送一个消息&＃xff0c;用这个生产者模拟线上的真实交易&＃xff0c;在实际生产中成交数据可以从日志中获取。

Kafka消费者&＃xff0c;SparkStream时实计算


object OrderConsumer
//Redis配置
val dbIndex &＃61; 0
//每件商品总销售额
val orderTotalKey &＃61; "app::order::total"
//每件商品上一分钟销售额
val oneMinTotalKey &＃61; "app::order::product"
//总销售额
val totalKey &＃61; "app::order::all"
def main(args: Array[String]): Unit &＃61;
// 创建 StreamingContext 时间片为1秒
val conf &＃61; new SparkConf().setMaster("local").setAppName("UserClickCountStat")
val ssc &＃61; new StreamingContext(conf, Seconds(1))
// Kafka 配置
val topics &＃61; Set("order")
val brokers &＃61; "127.0.0.1:9092"
val kafkaParams &＃61; Map[String, String](
"metadata.broker.list" -> brokers,
"serializer.class" -> "kafka.serializer.StringEncoder")
// 创建一个 direct stream
val kafkaStream &＃61; KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
//解析JSON
val events &＃61; kafkaStream.flatMap(line &＃61;> Some(JSON.parseObject(line._2)))
// 按ID分组统计个数与价格总合
val orders &＃61; events.map(x &＃61;> (x.getString("id"), x.getLong("price"))).groupByKey().map(x &＃61;> (x._1, x._2.size, x._2.reduceLeft(_ &＃43; _)))
//输出
orders.foreachRDD(x &＃61;>
x.foreachPartition(partition &＃61;>
partition.foreach(x &＃61;>
println("id&＃61;" &＃43; x._1 &＃43; " count&＃61;" &＃43; x._2 &＃43; " price&＃61;" &＃43; x._3)
//保存到Redis中
val jedis &＃61; RedisClient.pool.getResource
jedis.select(dbIndex)
//每个商品销售额累加
jedis.hincrBy(orderTotalKey, x._1, x._3)
//上一分钟第每个商品销售额
jedis.hset(oneMinTotalKey, x._1.toString, x._3.toString)
//总销售额累加
jedis.incrBy(totalKey, x._3)
RedisClient.pool.returnResource(jedis)
)
))
ssc.start()
ssc.awaitTermination()



消费者每分钟输出

id&＃61;4 count&＃61;3 price&＃61;7208
id&＃61;8 count&＃61;2 price&＃61;10152
id&＃61;7 count&＃61;1 price&＃61;6928
id&＃61;5 count&＃61;1 price&＃61;3327
id&＃61;6 count&＃61;3 price&＃61;20483
id&＃61;0 count&＃61;2 price&＃61;9882
id&＃61;2 count&＃61;2 price&＃61;9191
id&＃61;3 count&＃61;2 price&＃61;8211
id&＃61;1 count&＃61;3 price&＃61;9906

Redis客户端

object RedisClient extends Serializable
val redisHost &＃61; "127.0.0.1"
val redisPort &＃61; 6379
val redisTimeout &＃61; 30000
lazy val pool &＃61; new JedisPool(new GenericObjectPoolConfig(), redisHost, redisPort, redisTimeout)
lazy val hook &＃61; new Thread
override def run &＃61;
println("Execute hook thread: " &＃43; this)
pool.destroy()


sys.addShutdownHook(hook.run)
def main(args: Array[String]): Unit &＃61;
val dbIndex &＃61; 0
val jedis &＃61; RedisClient.pool.getResource
jedis.select(dbIndex)
jedis.set("test", "1")
println(jedis.get("test"))
RedisClient.pool.returnResource(jedis)



Redis结果

上一分钟商品销售额&＃xff0c;有了这个数据就可以做成动态的图表展示时实交易额了

每件商品总销售额

总销售额&＃xff0c;这就是天猫大屏上的1111亿了

完整代码地址

http://git.oschina.net/whzhaochao/spark-learning/tree/master/spark/src/main/scala/com/spark/stream/order

原文地址&＃xff1a;http://blog.csdn.net/whzhaochao/article/details/77717660